package com.sali.service.data_processor;


import com.alibaba.fastjson.JSONArray;
import com.alibaba.fastjson.JSONObject;
import com.sali.commons.utils.JSONFileReader;
import com.sali.entity.ProcessDefinition;
import com.sali.entity.TransferObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

// 数据处理的总入口
public class DataProcessor {

    public static final Logger log = LoggerFactory.getLogger(DataProcessor.class);

    // 保留协议类型和处理流程之间的映射关系
    private Map<String, DataProcessContext> protocolType2ProcessContext = new HashMap<>();

    public DataProcessor(String processConfigPath) {
        //1、初始化数据处理流程
        initDataProcessContext(processConfigPath);
    }

    private void initDataProcessContext(String processConfigPath) {
        try {
            JSONObject configContent = JSONFileReader.readJSONObjectFromFile(processConfigPath);
            if (configContent == null) {
                throw new RuntimeException("读取处理流程配置" + processConfigPath + "失败");
            }

            // 加载默认配置
            JSONObject defaultContent = configContent.getJSONObject("default");
            ProcessDefinition defaultProcessConfig = JSONObject.parseObject(defaultContent.toJSONString(), ProcessDefinition.class);
            log.info("start loading [default] config...");
            this.protocolType2ProcessContext.put("default", new DataProcessContext(defaultProcessConfig));
            log.info("end loading [default] config end...");

            // 加载自定义配置
            JSONArray customizedContents = configContent.getJSONArray("customized");
            if (customizedContents != null) {
                for (int i = 0; i < customizedContents.size(); i++) {
                    ProcessDefinition customizedProcessConfig = customizedContents.getObject(i, ProcessDefinition.class);
                    List<String> protocolTypes = customizedProcessConfig.getProtocolType();
                    if (protocolTypes != null) {
                        log.info("start loading [{}] config...", customizedProcessConfig.getName());
                        DataProcessContext cdpc = new DataProcessContext(customizedProcessConfig);
                        log.info("end loading [{}] config...", customizedProcessConfig.getName());
                        for (String protocolType : protocolTypes) {
                            this.protocolType2ProcessContext.put(protocolType, cdpc);

                        }
                    }
                }
            }
        } catch (Exception e) {
            log.error("初始化数据处理链路失败：{}", e.getMessage(), e);
        }
    }

    // 将数据提交到对应的处理流程中
    public void submit(TransferObject obj) {

        try {
            String protocolType = obj.getProtocolCode();

            DataProcessContext dpc = null;
            if (protocolType == null || !this.protocolType2ProcessContext.containsKey(protocolType)) {
                protocolType = "default";
            }
            dpc = this.protocolType2ProcessContext.get(protocolType);
            // 提交数据到disruptor
            dpc.put(obj);
        } catch (Exception e) {
            log.error("[data-processor] 提交数据异常: {}", e.getMessage(), e);
        }

    }

}
